package com.ls.fw.data.hbase.support;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.function.Consumer;

import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.client.Append;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Increment;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Row;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.client.coprocessor.Batch;
import org.apache.hadoop.hbase.util.Bytes;

import com.ls.fw.data.hbase.exception.HbaseException;
import com.ls.fw.data.hbase.exception.YHBaseException;

/**
 * 
 * @author ls
 */
public class HBaseTable {
	private final Table table;
	private final TableName tableName;
	private final Model model;
	private boolean closed = false;

	public HBaseTable(TableName tableName,
	                     Table table) {
		this.tableName = tableName;
		this.table = table;
		this.model = new Model();
	}

	public <R extends HTableModel> Model<R> model() {
		return model;
	}

	public void close() {
		try {
			table().close();
		} catch (Throwable throwable) {
			throw convertedException(throwable);
		} finally {
			closed = true;
		}
	}

	private Table table() {
		return table;
	}
	
	public HTableDescriptor createTable(String[] familys)
			throws HbaseException {
		HTableDescriptor tableDesc = new HTableDescriptor(tableName.get());
		int size = 0 ;
		try {
			if(familys!=null){
				for (String string : familys) {
					tableDesc.addFamily(new HColumnDescriptor(string));
					size++;
				}
			}
		} catch (Exception e) {
			throw new HbaseException(e.getMessage(),e);
		}finally{
			if(size <= 0){
				throw new HbaseException("列簇不能为空!");
			}
		}
		return tableDesc;
	}
	
	private YHBaseException convertedException(Throwable throwable) {
		if (throwable instanceof YHBaseException) {
			return (YHBaseException) throwable;
		} else {
			return new YHBaseException(throwable);
		}
	}

	public boolean isClosed() {
		return closed;
	}

	public boolean exists(Get get) {
		try {
			return table().exists(get);
		} catch (Throwable throwable) {
			throw convertedException(throwable);
		}
	}

	public boolean[] exists(List<Get> gets) {
		try {
			return table().existsAll(gets);
		} catch (Throwable throwable) {
			throw convertedException(throwable);
		}
	}

	public Result append(Append append) {
		try {
			return table().append(append);
		} catch (Throwable throwable) {
			throw convertedException(throwable);
		}
	}

	public TableName tableName() {
		return tableName;
	}

	public Result increment(Increment increment) {
		try {
			return table().increment(increment);
		} catch (Throwable throwable) {
			throw convertedException(throwable);
		}
	}

	public Result get(Get get) {
		try {
			return table().get(get);
		} catch (Throwable throwable) {
			throw convertedException(throwable);
		}
	}

	public Result[] get(List<Get> gets) {
		try {
			return table().get(gets);
		} catch (Throwable throwable) {
			throw convertedException(throwable);
		}
	}

	public void put(Put put) {
		try {
			table().put(put);
		} catch (Throwable throwable) {
			throw convertedException(throwable);
		}
	}

	public void put(List<Put> puts) {
		try {
			table().put(puts);
		} catch (Throwable throwable) {
			throw convertedException(throwable);
		}
	}

	public void delete(String rowkey) {
	     delete(new Delete(Bytes.toBytes(rowkey)));
	}

	public void delete(String... rowkey) {
		List<Delete> list = new ArrayList<Delete>();
		if(rowkey != null && rowkey.length>0){
			for (String id : rowkey) {
				list.add(new Delete(Bytes.toBytes(id)));
			}
		}
		if(!list.isEmpty()){
			delete(list);
		}
	}


	public void delete(Delete delete) {
		try {
			table().delete(delete);
		} catch (Throwable throwable) {
			throw convertedException(throwable);
		}
	}

	public void delete(List<Delete> deletes) {
		try {
			table().delete(deletes);
		} catch (Throwable throwable) {
			throw convertedException(throwable);
		}
	}

	public ResultScanner scan(Scan scan) {
		try {
			return table().getScanner(scan);
		} catch (Throwable throwable) {
			throw convertedException(throwable);
		}
	}
	public void scan(Scan scan, RowHandler rowHandler) {
		ResultScanner rs = null;
		try {
			rs = table().getScanner(scan);
			for (Result r = rs.next(); r != null; r = rs.next()) {
				if(!rowHandler.handler(r)){
					break;
				}
			}
			rowHandler.complete();
		} catch (Throwable throwable) {
			rowHandler.catchException(throwable);
			throw convertedException(throwable);
		} finally{
			if(null != rs){
				rs.close();
			}
		}
	}

	public void batch(List<? extends Row> actions, Object[] results) {
		try {
			table().batch(actions, results);
		} catch (Throwable throwable) {
			throw convertedException(throwable);
		}
	}

	public <R> void batchCallback(List<? extends Row> actions,
	                              Object[] results,
	                              Batch.Callback<R> callback) {
		try {
			table().batchCallback(actions, results, callback);
		} catch (Throwable throwable) {
			throw convertedException(throwable);
		}
	}

	public class Model<R extends HTableModel> {

		public R append(Append append) {
			try {
				return convert(table().append(append));
			} catch (Throwable throwable) {
				throw convertedException(throwable);
			}
		}

		private R convert(Result result) {
			return HTableModel.newWrappedModel(tableName, result);
		}

		public R increment(Increment increment) {
			try {
				return convert(table().increment(increment));
			} catch (Throwable throwable) {
				throw convertedException(throwable);
			}
		}

		public R get(Get get) {
			try {
				return convert(table().get(get));
			} catch (Throwable throwable) {
				throw convertedException(throwable);
			}
		}

		public List<R> get(List<Get> gets) {
			try {
				return Arrays.asList(table().get(gets))
					.stream()
					.collect(ArrayList::new,
						(list, get) -> list.add(convert(get)),
						(list1, list2) -> list1.addAll(list2));

			} catch (Throwable throwable) {
				throw convertedException(throwable);
			}
		}

		public ReturnScanner<R> scan(Scan scan) {
			try {
				return new ReturnScanner(table().getScanner(scan));
			} catch (Throwable throwable) {
				throw convertedException(throwable);
			}
		}

		public class ReturnScanner<R extends HTableModel> {
			private ResultScanner scanner;

			public ReturnScanner(ResultScanner scanner) {
				this.scanner = scanner;
			}

			public R next() throws IOException {
				return convert(scanner.next());
			}

			private R convert(Result result) {
				return HTableModel.newWrappedModel(tableName, result);
			}

			public List<R> next(int nbRows) throws IOException {
				return Arrays.asList(scanner.next(nbRows))
					.stream()
					.collect(ArrayList::new,
						(list, get) -> list.add(convert(get)),
						(list1, list2) -> list1.addAll(list2));
			}

			public void close() {
				scanner.close();
			}

			public void forEach(Consumer<? super R> action) {
				scanner.forEach(result -> action.accept(this.convert(result)));
			}
		}
	}
	
	public static interface RowHandler{
		public boolean handler(Result result);
		public void catchException(Throwable throwable);
		public void complete();
	}
}
